
package com.ibm.datamodel.timeseries;

import java.util.Collection;
import java.util.HashMap;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Preconditions;
import com.google.gson.JsonArray;
import com.google.gson.JsonParser;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.Constants;
import com.ibm.util.DataTypes;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

/**
 * @author Yaoliang Chen
 * @version Jun 12, 2014 TODO P1 add version support TODO P3 a little wierd for
 *          the combination of get and bind in the same class
 */
public class TimeSeriesSchema implements ITimeSeriesSchema {
	// public static TimeSeriesSchema ARRAY_SCHEMA = new TimeSeriesSchema(
	// null,
	// (ArrayNode) JSONUtility
	// .fromStringWithNoException("[ {\"name\":\"payload(0)\", \"datatype\":
	// \"float\"}, "
	// + "{\"name\":\"payload(1)\", \"datatype\": \"float\"},"
	// + "{\"name\":\"payload(2)\", \"datatype\": \"float\"},"
	// + "{\"name\":\"payload(3)\", \"datatype\": \"float\"},"
	// + " {\"name\":\"payload(4)\", \"datatype\": \"float\"}] "),
	// null, (new DateTime()).getMillis());

	private long version;

	@Override
	public int compareTo(ITimeSeriesSchema o) {
		return this.schemaId.compareTo(o.getSchemaID());
	}

	private String tenantid;
	private String schemaId;
	byte[] dataTypes;
	protected HashMap<String, Integer> name2Index;
	ArrayNode tags;
	String[] tagNames;
	protected String tsKeyTagName;
	protected String timestampTagName;
	protected int tsKeyTagIndex;
	protected int timestampTagIndex;

	public long getVersion() {
		return version;
	}

	public byte getMode(byte[] data) {
		if (data == null || data.length < 7)
			return Constants.DATA_FLAG_INVALID;
		return data[0];
	}

	/*
	 * this function is to inherit parentschema's timestamp field and key field
	 */
	public static TimeSeriesSchema inheritSchema(TimeSeriesSchema parentschema,
			String tenantid, String schemaId, ArrayNode tags, long version) throws CPSException{
		ArrayNode newtags = inheritNode(parentschema.getTags().deepCopy());
		newtags.addAll(tags);
		return new TimeSeriesSchema(tenantid, schemaId, tags, version);
	}

	/*
	 * find the tskey and timestamp field in parent tags
	 */
	public static ArrayNode inheritNode(ArrayNode parent){
		ArrayNode subNode = JSONUtility.newArrayNode();
		for (JsonNode node : subNode) {
			ObjectNode _tmp = (ObjectNode) node;
			if (_tmp.has(MessageFields.ISKEY) && _tmp.get(MessageFields.ISKEY).asBoolean() == true) {
				subNode.add(node);
			}
			if (_tmp.has(MessageFields.ISTIME) && _tmp.get(MessageFields.ISTIME).asBoolean() == true) {
				subNode.add(node);
			}
		}
		return subNode;
	}



	public TimeSeriesSchema(String tenantid, String schemaId, ArrayNode tags, long version) throws CPSException {
		this.tsKeyTagName = MessageFields.TSKEY;
		this.tsKeyTagIndex = -1;
		this.timestampTagName = MessageFields.TIMESTAMP;
		this.timestampTagIndex = -1;
		this.version = version;
		this.schemaId = schemaId;
		this.name2Index = new HashMap<String, Integer>();
		if (tags == null) {
			return;
		}
		int len = tags.size();
		/* TODO tsKeyTagIndex & timestampTagIndex */
		// to search the tsKey node and timestamp node
		for (int i = 0; i < tags.size(); i++) {
			JsonNode v = tags.get(i);
			if (v.isObject() && v.has(MessageFields.ISKEY) && v.get(MessageFields.ISKEY).asBoolean(false)) {
				if (tsKeyTagIndex >= 0) {
					throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "only one ts key tag is allowed.");
				}
//				// tskey don't have a datatype
//				if (v.has(MessageFields.DATATYPE)) {
//					throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "cannot specify data type for ts key.");
//				}
				if (!v.has(MessageFields.NAME)) {
					throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "Please specify tag name for ts key.");
				}

				tsKeyTagName = v.get(MessageFields.NAME).asText(MessageFields.TSKEY);
				tsKeyTagIndex = i;

			} else if (v.isObject() && v.has(MessageFields.ISTIME) && v.get(MessageFields.ISTIME).asBoolean(false)) {
				if (timestampTagIndex >= 0) {
					throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "only one timestamp tag is allowed.");
				}
				if (v.has(MessageFields.DATATYPE)) {
					byte dt = DataTypes.getDataType(v.get(MessageFields.DATATYPE).asText());
					if (dt != DataTypes.DT_LONG && dt != DataTypes.DT_STRING) {
						throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "cannot specify data type for time stamp.");
					}
				} else {
					((ObjectNode) v).put(MessageFields.DATATYPE, DataTypes.getDataType(DataTypes.DT_STRING));
				}

				if (!v.has(MessageFields.NAME)) {
					throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "Please specify tag name for time stamp.");
				}
				timestampTagName = v.get(MessageFields.NAME).asText(MessageFields.TIMESTAMP);
				timestampTagIndex = i;
			}
		}
		boolean hasKey = true;

		if (timestampTagIndex == -1) {
			timestampTagIndex = len++;
			ObjectNode node = JSONUtility.newObjectNode(MessageFields.NAME, timestampTagName);
			node.put(MessageFields.DATATYPE, "long");
			node.put(MessageFields.ISKEY, true);
			tags.add(node);
		}

		if (tsKeyTagIndex == -1) {
			hasKey = false;
			tsKeyTagIndex = len++;
			ObjectNode node = JSONUtility.newObjectNode(MessageFields.NAME, tsKeyTagName);
			node.put(MessageFields.DATATYPE, "long");
			node.put(MessageFields.ISTIME, true);
			tags.add(node);
		}

		this.dataTypes = new byte[len];
		this.tenantid = tenantid;
		this.tags = tags;
		this.tagNames = new String[len];

//		int i = 0;
		// add tskey node and timestamp node first
		if (tsKeyTagIndex == -1) {
//			name2Index.put(tsKeyTagName, i);
//			tagNames[i] = tsKeyTagName;
//			dataTypes[i] = DataTypes.DT_LONG;
//			tsKeyTagIndex = (i++);

		}

		if (timestampTagIndex == -1) {
//			name2Index.put(timestampTagName, i);
//			tagNames[i] = timestampTagName;
//			this.dataTypes[i] = DataTypes.DT_LONG;
//			timestampTagIndex = 0;

		}

		int tagLength = tags.size();
		for (int j = 0; j < tagLength; j++) {
			JsonNode v = tags.get(j);
			String tag = null;
			byte datatype = 0;
			if (v.isObject()) {
				if (!v.has(MessageFields.NAME)) {
					throw new CPSException(ErrorCode.NULL_FIELD_ERROR,
							"Please provide the tag name instead of " + v.toString());
				}
				tag = (String) v.get(MessageFields.NAME).asText();
				if (v.has(MessageFields.ISKEY) && v.get(MessageFields.ISKEY).asBoolean(false)) {
					datatype = DataTypes.DT_LONG;
				} else {
					datatype = v.has(MessageFields.DATATYPE)
							? DataTypes.getDataType((String) v.get(MessageFields.DATATYPE).asText())
							: DataTypes.DT_DEFAULT;
				}

			} else {
				tag = v.asText();
				datatype = DataTypes.DT_DEFAULT;
			}
			name2Index.put(tag, j);
			tagNames[j] = tag;
//			System.out.println(v.toString());
			Preconditions.checkArgument(datatype > 0, "can not parse data type tag[" + datatype + "] while tag is " + tag);
			this.dataTypes[j] = datatype;
		}
	}

	/**
	 * whether the tenant can access to the current time series type
	 *
	 * @param tenant
	 *            tenant name
	 * @return
	 */
	public boolean isTimeSeriesAccessible(String tenant) {
		if (this.tenantid == null) {
			return true;
		}
		return this.tenantid.equals(tenant);
	}

	/*
	 * (non-Javadoc)
	 *
	 * @see com.WhaleDB.datamodel.timeseries.ITimeSeriesTags#getdataTypes()
	 */
	@Override
	public byte[] getDataTypes() {
		return dataTypes;
	}

	public byte getDataType(int index) {
		if (index >= 0) {
			return dataTypes[index];
		}
		else {
			int[] indexs = parseTagId(index);
			int id = indexs[0];
			int subindex = indexs[1];
			switch (dataTypes[id]) {
			case DataTypes.DT_DOUBLE_ARRAY:
				return DataTypes.DT_DOUBLE;
			case DataTypes.DT_INT_ARRAY:
				return DataTypes.DT_LONG;
			case DataTypes.DT_STRING_ARRAY:
				return DataTypes.DT_STRING;
			default:
				// this may be not true
				return DataTypes.DT_DEFAULT;
			}
		}
	}

	/**
	 * get the ArrayNode of the schema of the tags
	 *
	 * @return
	 */
	public ArrayNode getTags() {
		return tags;
	}

	/*
	 * (non-Javadoc)
	 *
	 * @see com.WhaleDB.datamodel.timeseries.ITimeSeriesTags#getAttributeSet()
	 */
	@Override
	public String[] getTagNames() {
		return tagNames;
	}

	/*
	 * (non-Javadoc)
	 *
	 * @see com.WhaleDB.datamodel.timeseries.ITimeSeriesTags#getAttributeCount()
	 */
	@Override
	public int getTagCount() {
		return dataTypes.length;
	}

	@Override
	public int getTagNumber(String name) {
		Integer tagId = name2Index.get(name);
		// modify by gxj to support array field.
		if (tagId == null) {
			// may input like this 'array(0)'
			String regex = "\\((\\d+)\\)";
			Pattern pattern = Pattern.compile(regex);
			Matcher m = pattern.matcher(name);
			if (m.find()) {
				String array = m.replaceAll("");

				tagId = name2Index.get(array);
				if (tagId != null) {
					return tagId;
				}
			}
			return -1;
		}
		return tagId;
	}

	/**
	 * get the index of the given tag, starting from 0
	 *
	 * @param name
	 * @return -1 indicates failure
	 */
	@Override
	public int getTagId(String name) {
		Integer tagId = name2Index.get(name);
		// modify by gxj to support array field.
		if (tagId == null) {
			// may input like this 'array(0)'
			String regex = "\\((\\d+)\\)";
			Pattern pattern = Pattern.compile(regex);
			Matcher m = pattern.matcher(name);
			if (m.find()) {
				String subtag = m.group(1);
				String array = m.replaceAll("");

				tagId = name2Index.get(array);
				if (tagId == null) {
					return -1;
				}

				int subindex = Integer.parseInt(subtag);
				if (subindex < 0) {
					return -1;
				}

				if (subindex >= (2 << 14) - 1 || tagId >= (2 << 16 - 1)) {
					// overflow
					return -1;
				}

				// we need to extract subtag and merge into a tagId.
				// here we use bit manipulation.
				// avoid generating -1.
				tagId = (2 << 30) | (subindex << 16) | tagId;
				return tagId;
			}
		}
		if (tagId == null) {
			return -1;
		}

		return tagId;
	}

	// this function is to parse tagid into bits when datatype is array.
	public static int[] parseTagId(int tagid) {
		int index = tagid << 16 >>> 16;
		int subindex = tagid << 2 >>> 18;
		return new int[] { index, subindex };
	}

	@Override
	public String getSchemaID() {
		return schemaId;
	}

	@Override
	public String getUserID() {
		return tenantid;
	}

	@Override
	public int getTimestampIndex() {
		return this.timestampTagIndex;
	}

	public String getSchemaJson() {
		ObjectNode objectNode = JSONUtility.newObjectNode();
		objectNode.put(MessageFields.TENANTID, tenantid);
		objectNode.put(MessageFields.TYPE, MessageFields.SCHEMA_TYPE);
		objectNode.put(MessageFields.SCHEMAID, schemaId);
		objectNode.set(MessageFields.SCHEMA, tags);
		return objectNode.toString();
	}

	public static String getCollectionSchemaJson(Collection<TimeSeriesSchema> schemas) {
		JsonArray arrayNode = new JsonArray();
		JsonParser parser = new JsonParser();
		for (TimeSeriesSchema schema : schemas) {
			arrayNode.add(parser.parse(schema.getSchemaJson()));
		}
		return arrayNode.toString();
	}

	@Override
	public int getKeyIndex() {
		return this.tsKeyTagIndex;
	}

	@Override
	public ParsedDataMessage buildParsedDataMessage(Comparable[] values) throws CPSException {
		ParsedDataMessage message = this.newMessage();
		for (int number = 0; number < values.length; number++) {
			if (values[number] == null) {
				message.setValue(number, null);
				if (number == this.getTimestampIndex() || number == this.getKeyIndex()) {
					throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
							"Timestamp or time series key shouldn't be null.");
				}
				continue;
			}
			try {
				switch (this.dataTypes[number]) {
				case DataTypes.DT_LONG:
					long value = (Long) values[number];
					message.setValue(number, value);
					if (number == this.getTimestampIndex()) {
						message.setTimestamp(value);
					} else if (number == this.getKeyIndex()) {
						message.setTsKey(value);
					}
					break;
				case DataTypes.DT_STRING:
					String str = (String) values[number];
					message.setValue(number, str);
					break;
				case DataTypes.DT_DOUBLE:
					message.setValue(number, (Double) values[number]);
					break;
				case DataTypes.DT_BOOLEAN:
					message.setValue(number, (Boolean) values[number]);
					break;
				default:
					throw new CPSException(ErrorCode.ERROR_MESSAGE_TYPE,
							"Error message type " + this.dataTypes[number]);
				}
			} catch (ClassCastException e) {
				throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
						"Expected: " + dataTypes[number] + ", actual " + values[number].getClass());
			}
		}

		return message;
	}

	@Override
	public String getKeyName() {
		return this.tsKeyTagName;
	}

	@Override
	public ParsedDataMessage newMessage() {
		return new ParsedDataMessage(this.tagNames.length);
	}

}
